package com.wuwangfu.join;

import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.java.tuple.Tuple3;
import org.apache.flink.api.java.tuple.Tuple6;
import org.apache.flink.streaming.api.TimeCharacteristic;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.co.ProcessJoinFunction;
import org.apache.flink.streaming.api.functions.timestamps.BoundedOutOfOrdernessTimestampExtractor;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.util.Collector;

/**
 * @Author jcshen
 * @Date 2023-02-27
 * @PackageName:com.wuwangfu.join
 * @ClassName: IntervalJoin
 * @Description:
 * @Version 1.0.0
 * <p>
 * https://nightlies.apache.org/flink/flink-docs-release-1.14/docs/dev/datastream/operators/joining/#interval-join
 */
public class IntervalJoin {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);
        env.setStreamTimeCharacteristic(TimeCharacteristic.EventTime);
        /**
         * 500,A,8888
         * 3000,A,8888
         * 5000,A,8888
         */
        DataStreamSource<String> left = env.socketTextStream("localhost", 8888);
        /**
         * 1000,A,9999
         * 4000,A,9999
         */
        DataStreamSource<String> right = env.socketTextStream("localhost", 9999);

        //生成watermark
        SingleOutputStreamOperator<String> leftWatermark = left.assignTimestampsAndWatermarks(new BoundedOutOfOrdernessTimestampExtractor<String>(Time.seconds(0)) {
            @Override
            public long extractTimestamp(String element) {
                return Long.parseLong(element.split(",")[0]);
            }
        });
        SingleOutputStreamOperator<String> rightWatermark = right.assignTimestampsAndWatermarks(new BoundedOutOfOrdernessTimestampExtractor<String>(Time.seconds(0)) {
            @Override
            public long extractTimestamp(String element) {
                return Long.parseLong(element.split(",")[0]);
            }
        });
        //map
        SingleOutputStreamOperator<Tuple3<Long, String, String>> leftmaped = leftWatermark.map(new MapFunction<String, Tuple3<Long, String, String>>() {
            @Override
            public Tuple3<Long, String, String> map(String value) throws Exception {
                String[] fields = value.split(",");
                return Tuple3.of(Long.parseLong(fields[0]), fields[1], fields[2]);
            }
        });
        SingleOutputStreamOperator<Tuple3<Long, String, String>> rightmapped = rightWatermark.map(new MapFunction<String, Tuple3<Long, String, String>>() {
            @Override
            public Tuple3<Long, String, String> map(String value) throws Exception {
                String[] fields = value.split(",");
                return Tuple3.of(Long.parseLong(fields[0]), fields[1], fields[2]);
            }
        });
        //intervale join
        leftmaped
                .keyBy(t -> t.f1)//指定第一个流分组KeySelector
                .intervalJoin(rightmapped.keyBy(t -> t.f1))//调用intervalJoin方法，并指定第二个流的分组KeySelector
                .between(Time.seconds(-1), Time.seconds(1))//设置join的时间区间范围为当前数据时间 ±1 秒
                .upperBoundExclusive()//默认join时间范围为前后都包含的闭区间，现在设置为前闭后开区间
                .process(new ProcessJoinFunction<
                        Tuple3<Long, String, String>, //第一个流数据（左流）输入的数据类型
                        Tuple3<Long, String, String>, //第二个流数据（右流）输入的类型
                        Tuple6<Long, String, String, Long, String, String>>() {//join后输出的数据类型
                    @Override
                    public void processElement(Tuple3<Long, String, String> left,//左流输入的一条数据
                                               Tuple3<Long, String, String> right,//右流输入的一条数据
                                               Context ctx, //上下文信息，可以获取各个流的timestamp和侧流输出的output
                                               //用来输出join上的数据的Collector
                                               Collector<Tuple6<Long, String, String, Long, String, String>> out) throws Exception {
                        //将join上的数据添加到Collector中输出
                        out.collect(Tuple6.of(left.f0, left.f1, left.f2, right.f0, right.f1, right.f2));

                    }
                }).print();

        env.execute();
    }
}
